package com.ruoyi.tzai.service.task.impl;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import com.ruoyi.tzai.domain.ApiKeyPool;
import com.ruoyi.tzai.service.IApiKeyPoolService;
import com.ruoyi.tzai.service.IFileUploadService;
import com.ruoyi.tzai.service.ai.AIServiceFactory;
import com.ruoyi.tzai.service.ai.IAIService;
import com.ruoyi.tzai.service.task.IParallelProcessorService;
import com.ruoyi.tzai.service.tts.ITTSService;
import com.ruoyi.tzai.service.tts.TTSServiceFactory;

/**
 * 并行处理器Service业务层处理
 * 
 * @author tzai
 * @date 2025-11-12
 */
@Service
public class ParallelProcessorServiceImpl implements IParallelProcessorService 
{
    private static final Logger log = LoggerFactory.getLogger(ParallelProcessorServiceImpl.class);

    @Autowired
    private IApiKeyPoolService apiKeyPoolService;

    @Autowired
    private AIServiceFactory aiServiceFactory;

    @Autowired
    private TTSServiceFactory ttsServiceFactory;

    @Autowired
    private IFileUploadService fileUploadService;

    // 翻译线程池
    private final Executor translationExecutor;
    
    // TTS线程池
    private final Executor ttsExecutor;

    public ParallelProcessorServiceImpl() {
        // 翻译线程池
        ThreadPoolTaskExecutor translationPool = new ThreadPoolTaskExecutor();
        translationPool.setCorePoolSize(5);
        translationPool.setMaxPoolSize(10);
        translationPool.setQueueCapacity(50);
        translationPool.setThreadNamePrefix("translation-");
        translationPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        translationPool.initialize();
        this.translationExecutor = translationPool;

        // TTS线程池
        ThreadPoolTaskExecutor ttsPool = new ThreadPoolTaskExecutor();
        ttsPool.setCorePoolSize(5);
        ttsPool.setMaxPoolSize(10);
        ttsPool.setQueueCapacity(50);
        ttsPool.setThreadNamePrefix("tts-");
        ttsPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        ttsPool.initialize();
        this.ttsExecutor = ttsPool;
    }

    /**
     * 并行翻译（跨模型）
     */
    @Override
    public String parallelTranslate(Long taskId, Long stepId, String text, 
                                    String sourceLang, List<Long> modelIds, int parallelLevel) 
    {
        log.info("开始并行翻译: taskId={}, parallelLevel={}", taskId, parallelLevel);
        
        try {
            // 1. 获取可用Key（跨模型）
            List<ApiKeyPool> keys = apiKeyPoolService.acquireKeysAcrossModels(modelIds, parallelLevel);
            
            if (keys.isEmpty()) {
                throw new RuntimeException("没有可用的AI Key");
            }
            
            log.info("获取到 {} 个AI Key，来自 {} 个模型", 
                     keys.size(), 
                     keys.stream().map(ApiKeyPool::getResourceId).distinct().count());
            
            // 2. 分片文本
            List<String> chunks = splitText(text, 1000);
            log.info("文本分成 {} 个chunk", chunks.size());
            
            // 3. 并行翻译
            List<CompletableFuture<ChunkResult>> futures = new ArrayList<>();
            
            for (int i = 0; i < chunks.size(); i++) {
                final int chunkIndex = i;
                final String chunk = chunks.get(i);
                final ApiKeyPool key = keys.get(i % keys.size());
                
                CompletableFuture<ChunkResult> future = CompletableFuture.supplyAsync(() -> {
                    long startTime = System.currentTimeMillis();
                    try {
                        log.debug("翻译Chunk {}: 使用Key {} (modelId={})", 
                                 chunkIndex, key.getKeyName(), key.getResourceId());
                        
                        // 创建AI服务
                        IAIService aiService = aiServiceFactory.getService(key.getResourceId());
                        if (aiService == null) {
                            throw new RuntimeException("无法创建AI服务: modelId=" + key.getResourceId());
                        }
                        
                        // 翻译
                        String targetLang = "cn".equals(sourceLang) ? "en" : "cn";
                        String translated = aiService.translate(chunk, sourceLang, targetLang);
                        
                        long responseTime = System.currentTimeMillis() - startTime;
                        
                        // 更新Key统计
                        apiKeyPoolService.updateKeyStats(key.getKeyId(), true, null, responseTime);
                        
                        log.debug("Chunk {} 翻译完成: {}ms", chunkIndex, responseTime);
                        
                        return new ChunkResult(chunkIndex, translated, key.getKeyId(), true, null);
                        
                    } catch (Exception e) {
                        long responseTime = System.currentTimeMillis() - startTime;
                        log.error("Chunk {} 翻译失败", chunkIndex, e);
                        
                        // 更新Key统计
                        apiKeyPoolService.updateKeyStats(key.getKeyId(), false, e.getMessage(), responseTime);
                        
                        return new ChunkResult(chunkIndex, null, key.getKeyId(), false, e.getMessage());
                        
                    } finally {
                        // 释放Key
                        apiKeyPoolService.releaseKey(key.getKeyId());
                    }
                }, translationExecutor);
                
                futures.add(future);
            }
            
            // 4. 等待所有翻译完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            
            // 5. 收集结果
            List<ChunkResult> results = futures.stream()
                .map(CompletableFuture::join)
                .sorted(Comparator.comparingInt(ChunkResult::getIndex))
                .collect(Collectors.toList());
            
            // 6. 检查失败
            List<ChunkResult> failures = results.stream()
                .filter(r -> !r.isSuccess())
                .collect(Collectors.toList());
            
            if (!failures.isEmpty()) {
                log.error("翻译失败: {} 个chunk失败", failures.size());
                throw new RuntimeException("翻译失败: " + failures.get(0).getError());
            }
            
            // 7. 合并结果
            String merged = results.stream()
                .map(ChunkResult::getText)
                .collect(Collectors.joining("\n"));
            
            log.info("并行翻译完成: {} 个chunk", chunks.size());
            
            return merged;
            
        } catch (Exception e) {
            log.error("并行翻译失败: taskId={}", taskId, e);
            throw new RuntimeException("并行翻译失败: " + e.getMessage(), e);
        }
    }

    /**
     * 并行TTS生成（同引擎）
     */
    @Override
    public List<AudioResult> parallelTTS(Long taskId, Long stepId, List<String> sentences,
                                         String language, Long ttsEngineId, 
                                         String voiceName, int parallelLevel) 
    {
        log.info("开始并行TTS: taskId={}, sentences={}, parallelLevel={}", 
                 taskId, sentences.size(), parallelLevel);
        
        try {
            // 1. 获取可用Key（同引擎）
            List<ApiKeyPool> keys = apiKeyPoolService.acquireKeys("tts_engine", ttsEngineId, parallelLevel);
            
            if (keys.isEmpty()) {
                throw new RuntimeException("没有可用的TTS Key: engineId=" + ttsEngineId);
            }
            
            log.info("获取到 {} 个TTS Key (engineId={})", keys.size(), ttsEngineId);
            
            // 2. 并行生成音频
            List<CompletableFuture<AudioResult>> futures = new ArrayList<>();
            
            for (int i = 0; i < sentences.size(); i++) {
                final int sentenceIndex = i;
                final String text = sentences.get(i);
                final ApiKeyPool key = keys.get(i % keys.size());
                
                CompletableFuture<AudioResult> future = CompletableFuture.supplyAsync(() -> {
                    long startTime = System.currentTimeMillis();
                    try {
                        log.debug("生成句子 {}: 使用Key {} (engineId={})", 
                                 sentenceIndex, key.getKeyName(), key.getResourceId());
                        
                        // 创建TTS服务
                        ITTSService ttsService = ttsServiceFactory.getService(ttsEngineId);
                        if (ttsService == null) {
                            throw new RuntimeException("无法创建TTS服务: engineId=" + ttsEngineId);
                        }
                        
                        // 设置API Key
                        ttsService.setApiKey(key.getApiKey());
                        
                        // 生成音频
                        ITTSService.TTSResult result = ttsService.generateSpeech(text, language, voiceName);
                        
                        // 上传到OSS
                        String ossKey = String.format("audio/temp/task_%d_%s_s%d.mp3", 
                                                     taskId, language, sentenceIndex);
                        String audioUrl = fileUploadService.uploadAudioToOSS(result.getAudioData(), ossKey);
                        
                        long responseTime = System.currentTimeMillis() - startTime;
                        
                        // 更新Key统计
                        apiKeyPoolService.updateKeyStats(key.getKeyId(), true, null, responseTime);
                        
                        log.debug("句子 {} TTS完成: {}ms, duration={}s", 
                                 sentenceIndex, responseTime, result.getDuration());
                        
                        return new AudioResult(sentenceIndex, audioUrl, ossKey, 
                                             result.getDuration(), key.getKeyId(), true, null);
                        
                    } catch (Exception e) {
                        long responseTime = System.currentTimeMillis() - startTime;
                        log.error("句子 {} TTS失败", sentenceIndex, e);
                        
                        // 更新Key统计
                        apiKeyPoolService.updateKeyStats(key.getKeyId(), false, e.getMessage(), responseTime);
                        
                        return new AudioResult(sentenceIndex, null, null, 0, 
                                             key.getKeyId(), false, e.getMessage());
                        
                    } finally {
                        // 释放Key
                        apiKeyPoolService.releaseKey(key.getKeyId());
                    }
                }, ttsExecutor);
                
                futures.add(future);
            }
            
            // 3. 等待所有音频生成完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            
            // 4. 收集结果
            List<AudioResult> results = futures.stream()
                .map(CompletableFuture::join)
                .sorted(Comparator.comparingInt(AudioResult::getIndex))
                .collect(Collectors.toList());
            
            // 5. 检查失败
            List<AudioResult> failures = results.stream()
                .filter(r -> !r.isSuccess())
                .collect(Collectors.toList());
            
            if (!failures.isEmpty()) {
                log.error("TTS失败: {} 个句子失败", failures.size());
                throw new RuntimeException("TTS失败: " + failures.get(0).getError());
            }
            
            log.info("并行TTS完成: {} 个句子", sentences.size());
            
            return results;
            
        } catch (Exception e) {
            log.error("并行TTS失败: taskId={}", taskId, e);
            throw new RuntimeException("并行TTS失败: " + e.getMessage(), e);
        }
    }

    /**
     * 分割文本为chunks
     */
    private List<String> splitText(String text, int chunkSize) {
        List<String> chunks = new ArrayList<>();
        
        if (text == null || text.isEmpty()) {
            return chunks;
        }
        
        // 按段落分割
        String[] paragraphs = text.split("\n+");
        
        StringBuilder currentChunk = new StringBuilder();
        
        for (String para : paragraphs) {
            // 如果当前chunk + 新段落超过限制，则保存当前chunk
            if (currentChunk.length() + para.length() > chunkSize && currentChunk.length() > 0) {
                chunks.add(currentChunk.toString());
                currentChunk = new StringBuilder();
            }
            
            // 如果单个段落超过限制，强制分割
            if (para.length() > chunkSize) {
                // 先保存当前chunk
                if (currentChunk.length() > 0) {
                    chunks.add(currentChunk.toString());
                    currentChunk = new StringBuilder();
                }
                
                // 强制分割长段落
                for (int i = 0; i < para.length(); i += chunkSize) {
                    chunks.add(para.substring(i, Math.min(i + chunkSize, para.length())));
                }
            } else {
                if (currentChunk.length() > 0) {
                    currentChunk.append("\n");
                }
                currentChunk.append(para);
            }
        }
        
        // 添加最后一个chunk
        if (currentChunk.length() > 0) {
            chunks.add(currentChunk.toString());
        }
        
        return chunks;
    }
}

